Skip to content

File Module

erniebot_agent.file

The file module provides a series of classes for managing files, which facilitate user interaction with the agent, including the File base class and its subclasses, FileManager, GlobalFileManagerHandler, and RemoteFileClient, which interacts with remote file servers.

It is recommended to use the GlobalFileManagerHandler to initialize the FileManager and obtain the global FileManager at the beginning of the event loop. Afterwards, you can simply use this global FileManager to perform operations such as adding, deleting, and querying files, as well as obtaining files generated by the agent.

A few notes about this submodule:

  • If you do not set environment variable AISTUDIO_ACCESS_TOKEN, it will be under default setting.

  • Method GlobalFileManagerHandler().configure() can only be called once at the beginning.

  • When you want to get a file manger, you can use method GlobalFileManagerHandler().get().

  • The lifecycle of the FileManager class is synchronized with the event loop.

  • FileManager class is Noncopyable.

  • If you want to get the content of File object, you can use read_contents and use write_contents_to create the file to location you want.

  • We do not recommend you to create File object yourself.

Examples:

>>> from erniebot_agent.file import GlobalFileManagerHandler
>>> async def demo_function():
>>>     # need to use at the beginning of event loop
>>>     await GlobalFileManagerHandler().configure(save_dir='your_path')
>>>     file_manager = await GlobalFileManagerHandler().get()
>>>     local_file = await file_manager.create_file_from_path(file_path='your_path', file_type='local')

>>>     file = file_manager.look_up_file_by_id(file_id='your_file_id')
>>>     file_content = await file.read_contents() # get file content(bytes)
>>>     await local_file.write_contents_to('your_willing_path') # save to location you want

erniebot_agent.file.base

File

Abstract base class representing a generic file.

Attributes:

Name Type Description
id str

Unique identifier for the file.
filename str

File name.
byte_size int

Size of the file in bytes.
created_at str

Timestamp indicating the file creation time.
purpose str

Purpose or use case of the file.
metadata Dict[str, Any]

Additional metadata associated with the file.

Methods:

Name Description
read_contents

Abstract method to asynchronously read the file contents.
write_contents_to

Asynchronously write the file contents to a local path.
get_file_repr

Return a string representation for use in specific contexts.
to_dict

Convert the File object to a dictionary.
Source code in erniebot-agent/src/erniebot_agent/file/base.py 
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
class File(metaclass=abc.ABCMeta):
    """
    Abstract base class representing a generic file.

    Attributes:
        id (str): Unique identifier for the file.
        filename (str): File name.
        byte_size (int): Size of the file in bytes.
        created_at (str): Timestamp indicating the file creation time.
        purpose (str): Purpose or use case of the file.
        metadata (Dict[str, Any]): Additional metadata associated with the file.

    Methods:
        read_contents: Abstract method to asynchronously read the file contents.
        write_contents_to: Asynchronously write the file contents to a local path.
        get_file_repr: Return a string representation for use in specific contexts.
        to_dict: Convert the File object to a dictionary.
    """

    def __init__(
        self,
        *,
        id: str,
        filename: str,
        byte_size: int,
        created_at: str,
        purpose: str,
        metadata: Dict[str, Any],
    ) -> None:
        """
        Init method for the File class.

        Args:
            id (str): Unique identifier for the file.
            filename (str): File name.
            byte_size (int): Size of the file in bytes.
            created_at (str): Timestamp indicating the file creation time.
            purpose (str): Purpose or use case of the file.
            metadata (Dict[str, Any]): Additional metadata associated with the file.

        Returns:
            None
        """
        super().__init__()
        self.id = id
        self.filename = filename
        self.byte_size = byte_size
        self.created_at = created_at
        self.purpose = purpose
        self.metadata = metadata
        self._param_names = ["id", "filename", "byte_size", "created_at", "purpose", "metadata"]

    def __eq__(self, other: object) -> bool:
        if isinstance(other, File):
            return self.id == other.id
        else:
            return NotImplemented

    def __repr__(self) -> str:
        attrs_str = self._get_attrs_str()
        return f"<{self.__class__.__name__} {attrs_str}>"

    @abc.abstractmethod
    async def read_contents(self) -> bytes:
        raise NotImplementedError

    async def write_contents_to(self, local_path: Union[str, os.PathLike]) -> None:
        """Create a file to the location you want."""
        contents = await self.read_contents()
        await anyio.Path(local_path).write_bytes(contents)

    def get_file_repr(self) -> str:
        return f"<file>{self.id}</file>"

    def to_dict(self) -> dict:
        return {k: getattr(self, k) for k in self._param_names}

    def _get_attrs_str(self) -> str:
        return ", ".join(
            [
                f"id: {repr(self.id)}",
                f"filename: {repr(self.filename)}",
                f"byte_size: {repr(self.byte_size)}",
                f"created_at: {repr(self.created_at)}",
                f"purpose: {repr(self.purpose)}",
                f"metadata: {repr(self.metadata)}",
            ]
        )

__init__ 

__init__(*, id: str, filename: str, byte_size: int, created_at: str, purpose: str, metadata: Dict[str, Any]) -> None

Init method for the File class.

Parameters:

Name Type Description Default
id str

Unique identifier for the file.

 required
filename str

File name.

 required
byte_size int

Size of the file in bytes.

 required
created_at str

Timestamp indicating the file creation time.

 required
purpose str

Purpose or use case of the file.

 required
metadata Dict[str, Any]

Additional metadata associated with the file.

 required

Returns:

Type Description
None

None
Source code in erniebot-agent/src/erniebot_agent/file/base.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def __init__(
    self,
    *,
    id: str,
    filename: str,
    byte_size: int,
    created_at: str,
    purpose: str,
    metadata: Dict[str, Any],
) -> None:
    """
    Init method for the File class.

    Args:
        id (str): Unique identifier for the file.
        filename (str): File name.
        byte_size (int): Size of the file in bytes.
        created_at (str): Timestamp indicating the file creation time.
        purpose (str): Purpose or use case of the file.
        metadata (Dict[str, Any]): Additional metadata associated with the file.

    Returns:
        None
    """
    super().__init__()
    self.id = id
    self.filename = filename
    self.byte_size = byte_size
    self.created_at = created_at
    self.purpose = purpose
    self.metadata = metadata
    self._param_names = ["id", "filename", "byte_size", "created_at", "purpose", "metadata"]

write_contents_to async 

write_contents_to(local_path: Union[str, os.PathLike]) -> None

Create a file to the location you want.

Source code in erniebot-agent/src/erniebot_agent/file/base.py 
88
89
90
91
async def write_contents_to(self, local_path: Union[str, os.PathLike]) -> None:
    """Create a file to the location you want."""
    contents = await self.read_contents()
    await anyio.Path(local_path).write_bytes(contents)

erniebot_agent.file.local_file

LocalFile

Represents a local file.

Attributes:

Name Type Description
id str

Unique identifier for the file.
filename str

File name.
byte_size int

Size of the file in bytes.
created_at str

Timestamp indicating the file creation time.
purpose str

Purpose or use case of the file, including "assistants" and "assistants_output".
metadata Dict[str, Any]

Additional metadata associated with the file.
path Path

The path to the local file.

Methods:

Name Description
read_contents

Asynchronously read the contents of the local file.
write_contents_to

Asynchronously write the file contents to a local path.
get_file_repr

Return a string representation for use in specific contexts.
Source code in erniebot-agent/src/erniebot_agent/file/local_file.py 
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
class LocalFile(File):
    """
    Represents a local file.

    Attributes:
        id (str): Unique identifier for the file.
        filename (str): File name.
        byte_size (int): Size of the file in bytes.
        created_at (str): Timestamp indicating the file creation time.
        purpose (str): Purpose or use case of the file,
                       including "assistants" and "assistants_output".
        metadata (Dict[str, Any]): Additional metadata associated with the file.
        path (pathlib.Path): The path to the local file.

    Methods:
        read_contents: Asynchronously read the contents of the local file.
        write_contents_to: Asynchronously write the file contents to a local path.
        get_file_repr: Return a string representation for use in specific contexts.

    """

    def __init__(
        self,
        *,
        id: str,
        filename: str,
        byte_size: int,
        created_at: str,
        purpose: protocol.FilePurpose,
        metadata: Dict[str, Any],
        path: pathlib.Path,
        validate_file_id: bool = True,
    ) -> None:
        """
        Initialize a LocalFile object.

        Args:
            id (str): The unique identifier for the file.
            filename (str): The name of the file.
            byte_size (int): The size of the file in bytes.
            created_at (str): The timestamp indicating the file creation time.
            purpose (protocol.FilePurpose): The purpose or use case of the file.
            metadata (Dict[str, Any]): Additional metadata associated with the file.
            path (pathlib.Path): The path to the local file.
            validate_file_id (bool): Flag to validate the file ID. Default is True.

        Raises:
            ValueError: If the file ID is invalid.

        """
        if validate_file_id:
            if not protocol.is_local_file_id(id):
                raise ValueError(f"Invalid file ID: {id}")
        if not protocol.is_valid_file_purpose(purpose):
            raise ValueError(f"Invalid file purpose: {purpose}")
        super().__init__(
            id=id,
            filename=filename,
            byte_size=byte_size,
            created_at=created_at,
            purpose=purpose,
            metadata=metadata,
        )
        self.path = path

    async def read_contents(self) -> bytes:
        """Asynchronously read the contents of the local file."""
        return await anyio.Path(self.path).read_bytes()

    def _get_attrs_str(self) -> str:
        attrs_str = super()._get_attrs_str()
        attrs_str += f", path: {repr(self.path)}"
        return attrs_str

__init__ 

__init__(*, id: str, filename: str, byte_size: int, created_at: str, purpose: protocol.FilePurpose, metadata: Dict[str, Any], path: pathlib.Path, validate_file_id: bool = True) -> None

Initialize a LocalFile object.

Parameters:

Name Type Description Default
id str

The unique identifier for the file.

 required
filename str

The name of the file.

 required
byte_size int

The size of the file in bytes.

 required
created_at str

The timestamp indicating the file creation time.

 required
purpose FilePurpose

The purpose or use case of the file.

 required
metadata Dict[str, Any]

Additional metadata associated with the file.

 required
path Path

The path to the local file.

 required
validate_file_id bool

Flag to validate the file ID. Default is True.

 True

Raises:

Type Description
ValueError

If the file ID is invalid.
Source code in erniebot-agent/src/erniebot_agent/file/local_file.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def __init__(
    self,
    *,
    id: str,
    filename: str,
    byte_size: int,
    created_at: str,
    purpose: protocol.FilePurpose,
    metadata: Dict[str, Any],
    path: pathlib.Path,
    validate_file_id: bool = True,
) -> None:
    """
    Initialize a LocalFile object.

    Args:
        id (str): The unique identifier for the file.
        filename (str): The name of the file.
        byte_size (int): The size of the file in bytes.
        created_at (str): The timestamp indicating the file creation time.
        purpose (protocol.FilePurpose): The purpose or use case of the file.
        metadata (Dict[str, Any]): Additional metadata associated with the file.
        path (pathlib.Path): The path to the local file.
        validate_file_id (bool): Flag to validate the file ID. Default is True.

    Raises:
        ValueError: If the file ID is invalid.

    """
    if validate_file_id:
        if not protocol.is_local_file_id(id):
            raise ValueError(f"Invalid file ID: {id}")
    if not protocol.is_valid_file_purpose(purpose):
        raise ValueError(f"Invalid file purpose: {purpose}")
    super().__init__(
        id=id,
        filename=filename,
        byte_size=byte_size,
        created_at=created_at,
        purpose=purpose,
        metadata=metadata,
    )
    self.path = path

read_contents async 

read_contents() -> bytes

Asynchronously read the contents of the local file.

Source code in erniebot-agent/src/erniebot_agent/file/local_file.py 
129
130
131
async def read_contents(self) -> bytes:
    """Asynchronously read the contents of the local file."""
    return await anyio.Path(self.path).read_bytes()

erniebot_agent.file.remote_file

RemoteFile

Represents a remote file.

Attributes:

Name Type Description
id str

Unique identifier for the file.
filename str

File name.
byte_size int

Size of the file in bytes.
created_at str

Timestamp indicating the file creation time.
purpose str

Purpose or use case of the file, including "assistants" and "assistants_output".
metadata Dict[str, Any]

Additional metadata associated with the file.
client RemoteFileClient

The client of remote file.

Methods:

Name Description
read_contents

Asynchronously read the contents of the local file.
write_contents_to

Asynchronously write the file contents to a local path.
get_file_repr

Return a string representation for use in specific contexts.
delete

Asynchronously delete the file from client.
create_temporary_url

Asynchronously create a temporary URL for the file.
Source code in erniebot-agent/src/erniebot_agent/file/remote_file.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
class RemoteFile(File):
    """
    Represents a remote file.

    Attributes:
        id (str): Unique identifier for the file.
        filename (str): File name.
        byte_size (int): Size of the file in bytes.
        created_at (str): Timestamp indicating the file creation time.
        purpose (str): Purpose or use case of the file,
                       including "assistants" and "assistants_output".
        metadata (Dict[str, Any]): Additional metadata associated with the file.
        client (RemoteFileClient): The client of remote file.

    Methods:
        read_contents: Asynchronously read the contents of the local file.
        write_contents_to: Asynchronously write the file contents to a local path.
        get_file_repr: Return a string representation for use in specific contexts.
        delete: Asynchronously delete the file from client.
        create_temporary_url: Asynchronously create a temporary URL for the file.

    """

    def __init__(
        self,
        *,
        id: str,
        filename: str,
        byte_size: int,
        created_at: str,
        purpose: protocol.FilePurpose,
        metadata: Dict[str, Any],
        client: "RemoteFileClient",
        validate_file_id: bool = True,
    ) -> None:
        if validate_file_id:
            if not protocol.is_remote_file_id(id):
                raise FileError(f"Invalid file ID: {id}")
        if not protocol.is_valid_file_purpose(purpose):
            raise ValueError(f"Invalid file purpose: {purpose}")
        super().__init__(
            id=id,
            filename=filename,
            byte_size=byte_size,
            created_at=created_at,
            purpose=purpose,
            metadata=metadata,
        )
        self._client = client

    @property
    def client(self) -> "RemoteFileClient":
        return self._client

    async def read_contents(self) -> bytes:
        file_contents = await self._client.retrieve_file_contents(self.id)
        return file_contents

    async def delete(self) -> None:
        await self._client.delete_file(self.id)

    async def create_temporary_url(self, expire_after: float = 600) -> str:
        """To create a temporary valid URL for the file."""
        return await self._client.create_temporary_url(self.id, expire_after)

    def get_file_repr_with_url(self, url: str) -> str:
        return f"{self.get_file_repr()}<url>{url}</url>"

create_temporary_url async 

create_temporary_url(expire_after: float = 600) -> str

To create a temporary valid URL for the file.

Source code in erniebot-agent/src/erniebot_agent/file/remote_file.py 
91
92
93
async def create_temporary_url(self, expire_after: float = 600) -> str:
    """To create a temporary valid URL for the file."""
    return await self._client.create_temporary_url(self.id, expire_after)

erniebot_agent.file.file_manager

FileManager

Manages files, providing methods for creating, retrieving, and listing files.

Attributes:

Name Type Description
remote_file_client(RemoteFileClient)

The remote file client.
save_dir Optional[FilePath]

Directory for saving local files.
closed

Whether the file manager is closed.

Methods:

Name Description
create_file_from_path

Create a file from a specified file path.
create_local_file_from_path

Create a local file from a file path.
create_remote_file_from_path

Create a remote file from a file path.
create_file_from_bytes

Create a file from bytes.
retrieve_remote_file_by_id

Retrieve a remote file by its ID.
look_up_file_by_id

Look up a file by its ID.
list_remote_files

List remote files.
Source code in erniebot-agent/src/erniebot_agent/file/file_manager.py 
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
@final
class FileManager(Closeable, Noncopyable):
    """
    Manages files, providing methods for creating, retrieving, and listing files.

    Attributes:
        remote_file_client(RemoteFileClient): The remote file client.
        save_dir (Optional[FilePath]): Directory for saving local files.
        closed: Whether the file manager is closed.

    Methods:
        create_file_from_path: Create a file from a specified file path.
        create_local_file_from_path: Create a local file from a file path.
        create_remote_file_from_path: Create a remote file from a file path.
        create_file_from_bytes: Create a file from bytes.
        retrieve_remote_file_by_id: Retrieve a remote file by its ID.
        look_up_file_by_id: Look up a file by its ID.
        list_remote_files: List remote files.

    """

    _temp_dir: Optional[tempfile.TemporaryDirectory] = None

    def __init__(
        self,
        remote_file_client: Optional[RemoteFileClient] = None,
        save_dir: Optional[FilePath] = None,
        *,
        prune_on_close: bool = True,
    ) -> None:
        """
        Initialize the FileManager object.

        Args:
            remote_file_client (Optional[RemoteFileClient]): The remote file client.
            prune_on_close (bool): Control whether to automatically clean up files
                                   that can be safely deleted when the object is closed.
            save_dir (Optional[FilePath]): Directory for saving local files.

        Returns:
            None

        """
        super().__init__()

        self._remote_file_client = remote_file_client
        if save_dir is not None:
            self._save_dir = pathlib.Path(save_dir)
        else:
            # This can be done lazily, but we need to be careful about race conditions.
            self._temp_dir = self._create_temp_dir()
            self._save_dir = pathlib.Path(self._temp_dir.name)
        self._prune_on_close = prune_on_close

        self._file_registry: FileRegistry[_File] = FileRegistry()
        self._fully_managed_files: Deque[_File] = deque()

        self._closed = False

    @property
    def closed(self):
        return self._closed

    async def __aenter__(self) -> Self:
        return self

    async def __aexit__(
        self,
        exc_type: Optional[Type[BaseException]] = None,
        exc_value: Optional[BaseException] = None,
        traceback: Optional[TracebackType] = None,
    ) -> None:
        await self.close()

    @overload
    async def create_file_from_path(
        self,
        file_path: FilePath,
        *,
        file_purpose: protocol.FilePurpose = ...,
        file_metadata: Optional[Dict[str, Any]] = ...,
        file_type: Literal["local"],
    ) -> LocalFile:
        ...

    @overload
    async def create_file_from_path(
        self,
        file_path: FilePath,
        *,
        file_purpose: protocol.FilePurpose = ...,
        file_metadata: Optional[Dict[str, Any]] = ...,
        file_type: Literal["remote"],
    ) -> RemoteFile:
        ...

    @overload
    async def create_file_from_path(
        self,
        file_path: FilePath,
        *,
        file_purpose: protocol.FilePurpose = ...,
        file_metadata: Optional[Dict[str, Any]] = ...,
        file_type: None = ...,
    ) -> _File:
        ...

    async def create_file_from_path(
        self,
        file_path: FilePath,
        *,
        file_purpose: protocol.FilePurpose = "assistants",
        file_metadata: Optional[Dict[str, Any]] = None,
        file_type: Optional[Literal["local", "remote"]] = None,
    ) -> _File:
        """
        Create a file from a specified file path.

        Args:
            file_path (FilePath): The path to the file.
            file_purpose (FilePurpose): The purpose or use case of the file,
                    including `assistant`: used for llm and `assistant_output`: used for output.
            file_metadata (Optional[Dict[str, Any]]): Additional metadata associated with the file.
            file_type (Optional[Literal["local", "remote"]]): The type of file ("local" or "remote").

        Returns:
            Union[LocalFile, RemoteFile]: The created file.

        Raises:
            ValueError: If an unsupported file type is provided.

        """
        self.ensure_not_closed()
        file: _File
        if file_type is None:
            file_type = self._get_default_file_type()
        if file_type == "local":
            file = await self.create_local_file_from_path(file_path, file_purpose, file_metadata)
        elif file_type == "remote":
            file = await self.create_remote_file_from_path(file_path, file_purpose, file_metadata)
        else:
            raise ValueError(f"Unsupported file type: {file_type}")
        return file

    async def create_local_file_from_path(
        self,
        file_path: FilePath,
        file_purpose: protocol.FilePurpose,
        file_metadata: Optional[Dict[str, Any]],
    ) -> LocalFile:
        """
        Create a local file from a local file path.

        Args:
            file_path (FilePath): The path to the file.
            file_purpose (FilePurpose): The purpose or use case of the file,
                    including `assistant`: used for llm and `assistant_output`: used for output.
            file_metadata (Optional[Dict[str, Any]]): Additional metadata associated with the file.

        Returns:
            LocalFile: The created local file.

        """
        file = await self._create_local_file_from_path(
            pathlib.Path(file_path),
            file_purpose,
            file_metadata or {},
        )
        self._file_registry.register_file(file)
        return file

    async def create_remote_file_from_path(
        self,
        file_path: FilePath,
        file_purpose: protocol.FilePurpose,
        file_metadata: Optional[Dict[str, Any]],
    ) -> RemoteFile:
        """
        Create a remote file from a file path and upload it to the client.

        Args:
            file_path (FilePath): The path to the file.
            file_purpose (FilePurpose): The purpose or use case of the file,
                    including `assistant`: used for llm and `assistant_output`: used for output.
            file_metadata (Optional[Dict[str, Any]]): Additional metadata associated with the file.

        Returns:
            RemoteFile: The created remote file.

        """
        file = await self._create_remote_file_from_path(
            pathlib.Path(file_path),
            file_purpose,
            file_metadata,
        )
        self._file_registry.register_file(file)
        self._fully_managed_files.append(file)
        return file

    @overload
    async def create_file_from_bytes(
        self,
        file_contents: bytes,
        filename: str,
        *,
        file_purpose: protocol.FilePurpose = ...,
        file_metadata: Optional[Dict[str, Any]] = ...,
        file_type: Literal["local"],
    ) -> LocalFile:
        ...

    @overload
    async def create_file_from_bytes(
        self,
        file_contents: bytes,
        filename: str,
        *,
        file_purpose: protocol.FilePurpose = ...,
        file_metadata: Optional[Dict[str, Any]] = ...,
        file_type: Literal["remote"],
    ) -> RemoteFile:
        ...

    @overload
    async def create_file_from_bytes(
        self,
        file_contents: bytes,
        filename: str,
        *,
        file_purpose: protocol.FilePurpose = ...,
        file_metadata: Optional[Dict[str, Any]] = ...,
        file_type: None = ...,
    ) -> _File:
        ...

    async def create_file_from_bytes(
        self,
        file_contents: bytes,
        filename: str,
        *,
        file_purpose: protocol.FilePurpose = "assistants",
        file_metadata: Optional[Dict[str, Any]] = None,
        file_type: Optional[Literal["local", "remote"]] = None,
    ) -> _File:
        """
        Create a file from bytes.

        Args:
            file_contents (bytes): The contents of the file.
            filename (str): The name of the file.
            file_purpose (FilePurpose): The purpose or use case of the file.
            file_metadata (Optional[Dict[str, Any]]): Additional metadata associated with the file.
            file_type (Optional[Literal["local", "remote"]]): The type of file ("local" or "remote").

        Returns:
            Union[LocalFile, RemoteFile]: The created file.

        """
        self.ensure_not_closed()
        if file_type is None:
            file_type = self._get_default_file_type()
        file_path = self._get_unique_file_path(
            prefix=pathlib.PurePath(filename).stem,
            suffix=pathlib.PurePath(filename).suffix,
        )
        async_file_path = anyio.Path(file_path)
        await async_file_path.touch()
        should_remove_file = True
        try:
            async with await async_file_path.open("wb") as f:
                await f.write(file_contents)
            file: _File
            if file_type == "local":
                file = await self._create_local_file_from_path(file_path, file_purpose, file_metadata)
                should_remove_file = False
            elif file_type == "remote":
                file = await self._create_remote_file_from_path(
                    file_path,
                    file_purpose,
                    file_metadata,
                )
            else:
                raise ValueError(f"Unsupported file type: {file_type}")
        finally:
            if should_remove_file:
                await async_file_path.unlink()
        self._file_registry.register_file(file)
        self._fully_managed_files.append(file)
        return file

    async def retrieve_remote_file_by_id(self, file_id: str) -> RemoteFile:
        """
        Retrieve a remote file by its ID.

        Args:
            file_id (str): The ID of the remote file.

        Returns:
            RemoteFile: The retrieved remote file.

        """
        self.ensure_not_closed()
        file = await self._get_remote_file_client().retrieve_file(file_id)
        self._file_registry.register_file(file)
        return file

    async def list_remote_files(self) -> List[RemoteFile]:
        self.ensure_not_closed()
        files = await self._get_remote_file_client().list_files()
        return files

    def look_up_file_by_id(self, file_id: str) -> _File:
        """
        Look up a file by its ID.

        Args:
            file_id (str): The ID of the file.

        Returns:
            file[File]: The looked-up file.

        Raises:
            FileError: If the file with the specified ID is not found.

        """
        self.ensure_not_closed()
        file = self._file_registry.look_up_file(file_id)
        if file is None:
            raise FileError(
                f"File with ID {repr(file_id)} not found. "
                "Please check if `file_id` is correct and the file is registered."
            )
        return file

    def list_registered_files(self) -> List[_File]:
        """
        List remote files.

        Returns:
            List[RemoteFile]: The list of remote files.

        """
        self.ensure_not_closed()
        return self._file_registry.list_files()

    async def prune(self) -> None:
        """Clean local cache of file manager."""
        while True:
            try:
                file = self._fully_managed_files.pop()
            except IndexError:
                break
            if isinstance(file, RemoteFile):
                # FIXME: Currently this is not supported.
                # await file.delete()
                pass
            elif isinstance(file, LocalFile):
                assert self._save_dir.resolve() in file.path.resolve().parents
                await anyio.Path(file.path).unlink()
            else:
                raise AssertionError("Unexpected file type")
            self._file_registry.unregister_file(file)

    async def close(self) -> None:
        """Delete the file manager and clean up its cache"""
        if not self._closed:
            if self._remote_file_client is not None:
                await self._remote_file_client.close()
            if self._prune_on_close:
                await self.prune()
            if self._temp_dir is not None:
                self._clean_up_temp_dir(self._temp_dir)
            self._closed = True

    def sniff_and_extract_files_from_list(self, list_: List[Any]) -> List[_File]:
        files: List[_File] = []
        for item in list_:
            if not isinstance(item, str):
                continue
            if protocol.is_file_id(item):
                file_id = item
                try:
                    file = self.look_up_file_by_id(file_id)
                except FileError as e:
                    raise FileError(f"An unregistered file with ID {repr(file_id)} was found.") from e
                files.append(file)
        return files

    def sniff_and_extract_files_from_text(self, text: str) -> List[_File]:
        file_ids = protocol.extract_file_ids(text)
        files: List[_File] = []
        for file_id in file_ids:
            if protocol.is_file_id(file_id):
                try:
                    file = self.look_up_file_by_id(file_id)
                except FileError as e:
                    raise FileError(f"An unregistered file with ID {repr(file_id)} was found.") from e
                files.append(file)
        return files

    async def _create_local_file_from_path(
        self,
        file_path: pathlib.Path,
        file_purpose: protocol.FilePurpose,
        file_metadata: Optional[Dict[str, Any]],
    ) -> LocalFile:
        return create_local_file_from_path(
            pathlib.Path(file_path),
            file_purpose,
            file_metadata or {},
        )

    async def _create_remote_file_from_path(
        self,
        file_path: pathlib.Path,
        file_purpose: protocol.FilePurpose,
        file_metadata: Optional[Dict[str, Any]],
    ) -> RemoteFile:
        file = await self._get_remote_file_client().upload_file(file_path, file_purpose, file_metadata or {})
        return file

    def _get_remote_file_client(self) -> RemoteFileClient:
        if self._remote_file_client is None:
            raise AttributeError("No remote file client is set.")
        else:
            return self._remote_file_client

    def _get_default_file_type(self) -> Literal["local", "remote"]:
        if self._remote_file_client is not None:
            return "remote"
        else:
            return "local"

    def _get_unique_file_path(
        self, prefix: Optional[str] = None, suffix: Optional[str] = None
    ) -> pathlib.Path:
        filename = f"{prefix or ''}{str(uuid.uuid4())}{suffix or ''}"
        file_path = self._save_dir / filename
        return file_path

    @staticmethod
    def _create_temp_dir() -> tempfile.TemporaryDirectory:
        temp_dir = tempfile.TemporaryDirectory()
        return temp_dir

    @staticmethod
    def _clean_up_temp_dir(temp_dir: tempfile.TemporaryDirectory) -> None:
        try:
            temp_dir.cleanup()
        except Exception as e:
            logger.warning("Failed to clean up temporary directory: %s", temp_dir.name, exc_info=e)

__init__ 

__init__(remote_file_client: Optional[RemoteFileClient] = None, save_dir: Optional[FilePath] = None, *, prune_on_close: bool = True) -> None

Initialize the FileManager object.

Parameters:

Name Type Description Default
remote_file_client Optional[RemoteFileClient]

The remote file client.

 None
prune_on_close bool

Control whether to automatically clean up files that can be safely deleted when the object is closed.

 True
save_dir Optional[FilePath]

Directory for saving local files.

 None

Returns:

Type Description
None

None
Source code in erniebot-agent/src/erniebot_agent/file/file_manager.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def __init__(
    self,
    remote_file_client: Optional[RemoteFileClient] = None,
    save_dir: Optional[FilePath] = None,
    *,
    prune_on_close: bool = True,
) -> None:
    """
    Initialize the FileManager object.

    Args:
        remote_file_client (Optional[RemoteFileClient]): The remote file client.
        prune_on_close (bool): Control whether to automatically clean up files
                               that can be safely deleted when the object is closed.
        save_dir (Optional[FilePath]): Directory for saving local files.

    Returns:
        None

    """
    super().__init__()

    self._remote_file_client = remote_file_client
    if save_dir is not None:
        self._save_dir = pathlib.Path(save_dir)
    else:
        # This can be done lazily, but we need to be careful about race conditions.
        self._temp_dir = self._create_temp_dir()
        self._save_dir = pathlib.Path(self._temp_dir.name)
    self._prune_on_close = prune_on_close

    self._file_registry: FileRegistry[_File] = FileRegistry()
    self._fully_managed_files: Deque[_File] = deque()

    self._closed = False

close async 

close() -> None

Delete the file manager and clean up its cache

Source code in erniebot-agent/src/erniebot_agent/file/file_manager.py 
414
415
416
417
418
419
420
421
422
423
async def close(self) -> None:
    """Delete the file manager and clean up its cache"""
    if not self._closed:
        if self._remote_file_client is not None:
            await self._remote_file_client.close()
        if self._prune_on_close:
            await self.prune()
        if self._temp_dir is not None:
            self._clean_up_temp_dir(self._temp_dir)
        self._closed = True

create_file_from_bytes async 

create_file_from_bytes(file_contents: bytes, filename: str, *, file_purpose: protocol.FilePurpose = 'assistants', file_metadata: Optional[Dict[str, Any]] = None, file_type: Optional[Literal['local', 'remote']] = None) -> _File

Create a file from bytes.

Parameters:

Name Type Description Default
file_contents bytes

The contents of the file.

 required
filename str

The name of the file.

 required
file_purpose FilePurpose

The purpose or use case of the file.

 'assistants'
file_metadata Optional[Dict[str, Any]]

Additional metadata associated with the file.

 None
file_type Optional[Literal['local', 'remote']]

The type of file ("local" or "remote").

 None

Returns:

Type Description
_File

Union[LocalFile, RemoteFile]: The created file.
Source code in erniebot-agent/src/erniebot_agent/file/file_manager.py 
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
async def create_file_from_bytes(
    self,
    file_contents: bytes,
    filename: str,
    *,
    file_purpose: protocol.FilePurpose = "assistants",
    file_metadata: Optional[Dict[str, Any]] = None,
    file_type: Optional[Literal["local", "remote"]] = None,
) -> _File:
    """
    Create a file from bytes.

    Args:
        file_contents (bytes): The contents of the file.
        filename (str): The name of the file.
        file_purpose (FilePurpose): The purpose or use case of the file.
        file_metadata (Optional[Dict[str, Any]]): Additional metadata associated with the file.
        file_type (Optional[Literal["local", "remote"]]): The type of file ("local" or "remote").

    Returns:
        Union[LocalFile, RemoteFile]: The created file.

    """
    self.ensure_not_closed()
    if file_type is None:
        file_type = self._get_default_file_type()
    file_path = self._get_unique_file_path(
        prefix=pathlib.PurePath(filename).stem,
        suffix=pathlib.PurePath(filename).suffix,
    )
    async_file_path = anyio.Path(file_path)
    await async_file_path.touch()
    should_remove_file = True
    try:
        async with await async_file_path.open("wb") as f:
            await f.write(file_contents)
        file: _File
        if file_type == "local":
            file = await self._create_local_file_from_path(file_path, file_purpose, file_metadata)
            should_remove_file = False
        elif file_type == "remote":
            file = await self._create_remote_file_from_path(
                file_path,
                file_purpose,
                file_metadata,
            )
        else:
            raise ValueError(f"Unsupported file type: {file_type}")
    finally:
        if should_remove_file:
            await async_file_path.unlink()
    self._file_registry.register_file(file)
    self._fully_managed_files.append(file)
    return file

create_file_from_path async 

create_file_from_path(file_path: FilePath, *, file_purpose: protocol.FilePurpose = 'assistants', file_metadata: Optional[Dict[str, Any]] = None, file_type: Optional[Literal['local', 'remote']] = None) -> _File

Create a file from a specified file path.

Parameters:

Name Type Description Default
file_path FilePath

The path to the file.

 required
file_purpose FilePurpose

The purpose or use case of the file, including assistant: used for llm and assistant_output: used for output.

 'assistants'
file_metadata Optional[Dict[str, Any]]

Additional metadata associated with the file.

 None
file_type Optional[Literal['local', 'remote']]

The type of file ("local" or "remote").

 None

Returns:

Type Description
_File

Union[LocalFile, RemoteFile]: The created file.

Raises:

Type Description
ValueError

If an unsupported file type is provided.
Source code in erniebot-agent/src/erniebot_agent/file/file_manager.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
async def create_file_from_path(
    self,
    file_path: FilePath,
    *,
    file_purpose: protocol.FilePurpose = "assistants",
    file_metadata: Optional[Dict[str, Any]] = None,
    file_type: Optional[Literal["local", "remote"]] = None,
) -> _File:
    """
    Create a file from a specified file path.

    Args:
        file_path (FilePath): The path to the file.
        file_purpose (FilePurpose): The purpose or use case of the file,
                including `assistant`: used for llm and `assistant_output`: used for output.
        file_metadata (Optional[Dict[str, Any]]): Additional metadata associated with the file.
        file_type (Optional[Literal["local", "remote"]]): The type of file ("local" or "remote").

    Returns:
        Union[LocalFile, RemoteFile]: The created file.

    Raises:
        ValueError: If an unsupported file type is provided.

    """
    self.ensure_not_closed()
    file: _File
    if file_type is None:
        file_type = self._get_default_file_type()
    if file_type == "local":
        file = await self.create_local_file_from_path(file_path, file_purpose, file_metadata)
    elif file_type == "remote":
        file = await self.create_remote_file_from_path(file_path, file_purpose, file_metadata)
    else:
        raise ValueError(f"Unsupported file type: {file_type}")
    return file

create_local_file_from_path async 

create_local_file_from_path(file_path: FilePath, file_purpose: protocol.FilePurpose, file_metadata: Optional[Dict[str, Any]]) -> LocalFile

Create a local file from a local file path.

Parameters:

Name Type Description Default
file_path FilePath

The path to the file.

 required
file_purpose FilePurpose

The purpose or use case of the file, including assistant: used for llm and assistant_output: used for output.

 required
file_metadata Optional[Dict[str, Any]]

Additional metadata associated with the file.

 required

Returns:

Name Type Description
LocalFile LocalFile

The created local file.
Source code in erniebot-agent/src/erniebot_agent/file/file_manager.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
async def create_local_file_from_path(
    self,
    file_path: FilePath,
    file_purpose: protocol.FilePurpose,
    file_metadata: Optional[Dict[str, Any]],
) -> LocalFile:
    """
    Create a local file from a local file path.

    Args:
        file_path (FilePath): The path to the file.
        file_purpose (FilePurpose): The purpose or use case of the file,
                including `assistant`: used for llm and `assistant_output`: used for output.
        file_metadata (Optional[Dict[str, Any]]): Additional metadata associated with the file.

    Returns:
        LocalFile: The created local file.

    """
    file = await self._create_local_file_from_path(
        pathlib.Path(file_path),
        file_purpose,
        file_metadata or {},
    )
    self._file_registry.register_file(file)
    return file

create_remote_file_from_path async 

create_remote_file_from_path(file_path: FilePath, file_purpose: protocol.FilePurpose, file_metadata: Optional[Dict[str, Any]]) -> RemoteFile

Create a remote file from a file path and upload it to the client.

Parameters:

Name Type Description Default
file_path FilePath

The path to the file.

 required
file_purpose FilePurpose

The purpose or use case of the file, including assistant: used for llm and assistant_output: used for output.

 required
file_metadata Optional[Dict[str, Any]]

Additional metadata associated with the file.

 required

Returns:

Name Type Description
RemoteFile RemoteFile

The created remote file.
Source code in erniebot-agent/src/erniebot_agent/file/file_manager.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
async def create_remote_file_from_path(
    self,
    file_path: FilePath,
    file_purpose: protocol.FilePurpose,
    file_metadata: Optional[Dict[str, Any]],
) -> RemoteFile:
    """
    Create a remote file from a file path and upload it to the client.

    Args:
        file_path (FilePath): The path to the file.
        file_purpose (FilePurpose): The purpose or use case of the file,
                including `assistant`: used for llm and `assistant_output`: used for output.
        file_metadata (Optional[Dict[str, Any]]): Additional metadata associated with the file.

    Returns:
        RemoteFile: The created remote file.

    """
    file = await self._create_remote_file_from_path(
        pathlib.Path(file_path),
        file_purpose,
        file_metadata,
    )
    self._file_registry.register_file(file)
    self._fully_managed_files.append(file)
    return file

list_registered_files 

list_registered_files() -> List[_File]

List remote files.

Returns:

Type Description
List[_File]

List[RemoteFile]: The list of remote files.
Source code in erniebot-agent/src/erniebot_agent/file/file_manager.py 
385
386
387
388
389
390
391
392
393
394
def list_registered_files(self) -> List[_File]:
    """
    List remote files.

    Returns:
        List[RemoteFile]: The list of remote files.

    """
    self.ensure_not_closed()
    return self._file_registry.list_files()

look_up_file_by_id 

look_up_file_by_id(file_id: str) -> _File

Look up a file by its ID.

Parameters:

Name Type Description Default
file_id str

The ID of the file.

 required

Returns:

Type Description
_File

file[File]: The looked-up file.

Raises:

Type Description
FileError

If the file with the specified ID is not found.
Source code in erniebot-agent/src/erniebot_agent/file/file_manager.py 
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
def look_up_file_by_id(self, file_id: str) -> _File:
    """
    Look up a file by its ID.

    Args:
        file_id (str): The ID of the file.

    Returns:
        file[File]: The looked-up file.

    Raises:
        FileError: If the file with the specified ID is not found.

    """
    self.ensure_not_closed()
    file = self._file_registry.look_up_file(file_id)
    if file is None:
        raise FileError(
            f"File with ID {repr(file_id)} not found. "
            "Please check if `file_id` is correct and the file is registered."
        )
    return file

prune async 

prune() -> None

Clean local cache of file manager.

Source code in erniebot-agent/src/erniebot_agent/file/file_manager.py 
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
async def prune(self) -> None:
    """Clean local cache of file manager."""
    while True:
        try:
            file = self._fully_managed_files.pop()
        except IndexError:
            break
        if isinstance(file, RemoteFile):
            # FIXME: Currently this is not supported.
            # await file.delete()
            pass
        elif isinstance(file, LocalFile):
            assert self._save_dir.resolve() in file.path.resolve().parents
            await anyio.Path(file.path).unlink()
        else:
            raise AssertionError("Unexpected file type")
        self._file_registry.unregister_file(file)

retrieve_remote_file_by_id async 

retrieve_remote_file_by_id(file_id: str) -> RemoteFile

Retrieve a remote file by its ID.

Parameters:

Name Type Description Default
file_id str

The ID of the remote file.

 required

Returns:

Name Type Description
RemoteFile RemoteFile

The retrieved remote file.
Source code in erniebot-agent/src/erniebot_agent/file/file_manager.py 
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
async def retrieve_remote_file_by_id(self, file_id: str) -> RemoteFile:
    """
    Retrieve a remote file by its ID.

    Args:
        file_id (str): The ID of the remote file.

    Returns:
        RemoteFile: The retrieved remote file.

    """
    self.ensure_not_closed()
    file = await self._get_remote_file_client().retrieve_file(file_id)
    self._file_registry.register_file(file)
    return file

erniebot_agent.file.global_file_manager_handler

GlobalFileManagerHandler

Singleton handler for managing the global FileManager instance.

This class provides a singleton instance for managing the global FileManager and allows for its configuration and retrieval.

Methods:

Name Description
get

Asynchronously retrieves the global FileManager instance.
configure

Asynchronously configures the global FileManager at the beginning of event loop.
set

Asynchronously sets the global FileManager explicitly.
Source code in erniebot-agent/src/erniebot_agent/file/global_file_manager_handler.py 
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
@final
class GlobalFileManagerHandler(object):
    """Singleton handler for managing the global FileManager instance.

    This class provides a singleton instance for managing the global FileManager
    and allows for its configuration and retrieval.


    Methods:
        get: Asynchronously retrieves the global FileManager instance.
        configure: Asynchronously configures the global FileManager
                   at the beginning of event loop.
        set: Asynchronously sets the global FileManager explicitly.

    """

    _lock: asyncio.Lock
    _file_manager: Optional[FileManager]

    def __new__(cls) -> Self:
        loop = asyncio.get_running_loop()
        if loop not in _registry:
            handler = super().__new__(cls)
            handler._lock = asyncio.Lock()
            handler._file_manager = None
            _registry[loop] = handler
        return _registry[loop]

    async def get(self) -> FileManager:
        """
        Retrieve the global FileManager instance.

        This method returns the existing global FileManager instance,
        creating one if it doesn't exist.


        Returns:
            FileManager: The global FileManager instance.

        """

        async with self._lock:
            if self._file_manager is None:
                self._file_manager = await self._create_default_file_manager(
                    access_token=None,
                    save_dir=None,
                    enable_remote_file=False,
                )
            return self._file_manager

    async def configure(
        self,
        *,
        access_token: Optional[str] = None,
        save_dir: Optional[str] = None,
        enable_remote_file: bool = False,
        **opts: Any,
    ) -> None:
        """
        Configure the global FileManager.

        This method configures the global FileManager with the provided parameters
        at the beginning of event loop.
        If the global FileManager is already set, it raises an error.

        Args:
            access_token (Optional[str]): The access token for remote file client.
            save_dir (Optional[str]): The directory for saving local files.
            enable_remote_file (bool): Whether to enable remote file.
            **opts (Any): Additional options for FileManager.

        Returns:
            None

        Raises:
            RuntimeError: If the global FileManager is already set.

        """
        async with self._lock:
            if self._file_manager is not None:
                self._raise_file_manager_already_set_error()
            self._file_manager = await self._create_default_file_manager(
                access_token=access_token,
                save_dir=save_dir,
                enable_remote_file=enable_remote_file,
                **opts,
            )

    async def set(self, file_manager: FileManager) -> None:
        """
        Set the global FileManager explicitly.

        This method sets the global FileManager instance explicitly.
        If the global FileManager is already set, it raises an error.

        Args:
            file_manager (FileManager): The FileManager instance to set as global.

        Returns:
            None

        Raises:
            RuntimeError: If the global FileManager is already set.
        """
        async with self._lock:
            if self._file_manager is not None:
                self._raise_file_manager_already_set_error()
            self._file_manager = file_manager

    async def _create_default_file_manager(
        self,
        access_token: Optional[str],
        save_dir: Optional[str],
        enable_remote_file: bool,
        **opts: Any,
    ) -> FileManager:
        """Create the default FileManager instance."""

        async def _close_file_manager():
            await file_manager.close()

        if access_token is None:
            access_token = C.get_global_access_token()
        if save_dir is None:
            save_dir = C.get_global_save_dir()
        remote_file_client = None
        if enable_remote_file:
            if access_token is None:
                raise RuntimeError("An access token must be provided to enable remote file management.")
            remote_file_client = AIStudioFileClient(access_token=access_token)
        file_manager = FileManager(remote_file_client, save_dir, **opts)
        asyncio_atexit.register(_close_file_manager)
        return file_manager

    def _raise_file_manager_already_set_error(self) -> NoReturn:
        raise RuntimeError(
            "The global file manager can only be set once."
            " The setup can be done explicitly by calling"
            " `GlobalFileManagerHandler.configure` or `GlobalFileManagerHandler.set`,"
            " or implicitly by calling `GlobalFileManagerHandler.get`."
        )

configure async 

configure(*, access_token: Optional[str] = None, save_dir: Optional[str] = None, enable_remote_file: bool = False, **opts: Any) -> None

Configure the global FileManager.

This method configures the global FileManager with the provided parameters at the beginning of event loop. If the global FileManager is already set, it raises an error.

Parameters:

Name Type Description Default
access_token Optional[str]

The access token for remote file client.

 None
save_dir Optional[str]

The directory for saving local files.

 None
enable_remote_file bool

Whether to enable remote file.

 False
**opts Any

Additional options for FileManager.

 {}

Returns:

Type Description
None

None

Raises:

Type Description
RuntimeError

If the global FileManager is already set.
Source code in erniebot-agent/src/erniebot_agent/file/global_file_manager_handler.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
async def configure(
    self,
    *,
    access_token: Optional[str] = None,
    save_dir: Optional[str] = None,
    enable_remote_file: bool = False,
    **opts: Any,
) -> None:
    """
    Configure the global FileManager.

    This method configures the global FileManager with the provided parameters
    at the beginning of event loop.
    If the global FileManager is already set, it raises an error.

    Args:
        access_token (Optional[str]): The access token for remote file client.
        save_dir (Optional[str]): The directory for saving local files.
        enable_remote_file (bool): Whether to enable remote file.
        **opts (Any): Additional options for FileManager.

    Returns:
        None

    Raises:
        RuntimeError: If the global FileManager is already set.

    """
    async with self._lock:
        if self._file_manager is not None:
            self._raise_file_manager_already_set_error()
        self._file_manager = await self._create_default_file_manager(
            access_token=access_token,
            save_dir=save_dir,
            enable_remote_file=enable_remote_file,
            **opts,
        )

get async 

get() -> FileManager

Retrieve the global FileManager instance.

This method returns the existing global FileManager instance, creating one if it doesn't exist.

Returns:

Name Type Description
FileManager FileManager

The global FileManager instance.
Source code in erniebot-agent/src/erniebot_agent/file/global_file_manager_handler.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
async def get(self) -> FileManager:
    """
    Retrieve the global FileManager instance.

    This method returns the existing global FileManager instance,
    creating one if it doesn't exist.


    Returns:
        FileManager: The global FileManager instance.

    """

    async with self._lock:
        if self._file_manager is None:
            self._file_manager = await self._create_default_file_manager(
                access_token=None,
                save_dir=None,
                enable_remote_file=False,
            )
        return self._file_manager

set async 

set(file_manager: FileManager) -> None

Set the global FileManager explicitly.

This method sets the global FileManager instance explicitly. If the global FileManager is already set, it raises an error.

Parameters:

Name Type Description Default
file_manager FileManager

The FileManager instance to set as global.

 required

Returns:

Type Description
None

None

Raises:

Type Description
RuntimeError

If the global FileManager is already set.
Source code in erniebot-agent/src/erniebot_agent/file/global_file_manager_handler.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
async def set(self, file_manager: FileManager) -> None:
    """
    Set the global FileManager explicitly.

    This method sets the global FileManager instance explicitly.
    If the global FileManager is already set, it raises an error.

    Args:
        file_manager (FileManager): The FileManager instance to set as global.

    Returns:
        None

    Raises:
        RuntimeError: If the global FileManager is already set.
    """
    async with self._lock:
        if self._file_manager is not None:
            self._raise_file_manager_already_set_error()
        self._file_manager = file_manager

erniebot_agent.file.remote_file

AIStudioFileClient

Recommended remote file client: AI Studio.

Methods:

Name Description
upload_file

Upload a file to AI Studio client.
retrieve_file

Retrieve information about a file from AI Studio.
retrieve_file_contents

Retrieve the contents of a file from AI Studio.
list_files

List files available in AI Studio.
delete_file

Delete a file in AI Studio client(#TODO: not supported now).
create_temporary_url

Create a temporary URL for accessing a file in AI Studio.
close

Close the AIStudioFileClient.
Source code in erniebot-agent/src/erniebot_agent/file/remote_file.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
class AIStudioFileClient(RemoteFileClient):
    """
    Recommended remote file client: AI Studio.

    Methods:
        upload_file: Upload a file to AI Studio client.
        retrieve_file: Retrieve information about a file from AI Studio.
        retrieve_file_contents: Retrieve the contents of a file from AI Studio.
        list_files: List files available in AI Studio.
        delete_file: Delete a file in AI Studio client(#TODO: not supported now).
        create_temporary_url: Create a temporary URL for accessing a file in AI Studio.
        close: Close the AIStudioFileClient.

    """

    _BASE_URL: ClassVar[str] = "https://sandbox-aistudio.baidu.com"
    _UPLOAD_ENDPOINT: ClassVar[str] = "/llm/lmapp/files"
    _RETRIEVE_ENDPOINT: ClassVar[str] = "/llm/lmapp/files/{file_id}"
    _RETRIEVE_CONTENTS_ENDPOINT: ClassVar[str] = "/llm/lmapp/files/{file_id}/content"
    _LIST_ENDPOINT: ClassVar[str] = "/llm/lmapp/files"

    def __init__(
        self, access_token: str, *, aiohttp_session: Optional[aiohttp.ClientSession] = None
    ) -> None:
        """
        Initialize the AIStudioFileClient.

        Args:
            access_token (str): The access token for AI Studio.
            aiohttp_session (Optional[aiohttp.ClientSession]): A custom aiohttp session (default is None).

        """
        super().__init__()
        self._access_token = access_token
        if aiohttp_session is None:
            aiohttp_session = self._create_aiohttp_session()
        self._session = aiohttp_session
        self._closed = False

    @property
    def closed(self) -> bool:
        return self._closed

    async def upload_file(
        self, file_path: pathlib.Path, file_purpose: protocol.FilePurpose, file_metadata: Dict[str, Any]
    ) -> RemoteFile:
        """Upload a file to AI Studio client."""
        self.ensure_not_closed()
        url = self._get_url(self._UPLOAD_ENDPOINT)
        headers: Dict[str, str] = {}
        headers.update(self._get_default_headers())
        with file_path.open("rb") as file:
            form_data = aiohttp.FormData()
            form_data.add_field("file", file, filename=file_path.name)
            form_data.add_field("purpose", file_purpose)
            form_data.add_field("meta", json.dumps(file_metadata))
            resp_bytes = await self._request(
                "POST",
                url,
                data=form_data,
                headers=headers,
                raise_for_status=True,
            )
        result = self._get_result_from_response_body(resp_bytes)
        return self._create_file_obj_from_dict(result)

    async def retrieve_file(self, file_id: str) -> RemoteFile:
        """Retrieve a file in AI Studio client by id."""
        self.ensure_not_closed()
        url = self._get_url(self._RETRIEVE_ENDPOINT).format(file_id=file_id)
        headers: Dict[str, str] = {}
        headers.update(self._get_default_headers())
        resp_bytes = await self._request(
            "GET",
            url,
            headers=headers,
            raise_for_status=True,
        )
        result = self._get_result_from_response_body(resp_bytes)
        return self._create_file_obj_from_dict(result)

    async def retrieve_file_contents(self, file_id: str) -> bytes:
        """Retrieve file content in AI Studio client by id."""
        self.ensure_not_closed()
        url = self._get_url(self._RETRIEVE_CONTENTS_ENDPOINT).format(file_id=file_id)
        headers: Dict[str, str] = {}
        headers.update(self._get_default_headers())
        resp_bytes = await self._request(
            "GET",
            url,
            headers=headers,
            raise_for_status=True,
        )
        return resp_bytes

    async def list_files(self) -> List[RemoteFile]:
        """List files in AI Studio client."""
        self.ensure_not_closed()
        url = self._get_url(self._LIST_ENDPOINT)
        headers: Dict[str, str] = {}
        headers.update(self._get_default_headers())
        resp_bytes = await self._request(
            "GET",
            url,
            headers=headers,
            raise_for_status=True,
        )
        result = self._get_result_from_response_body(resp_bytes)
        files: List[RemoteFile] = []
        for item in result:
            file = self._create_file_obj_from_dict(item)
            files.append(file)
        return files

    async def delete_file(self, file_id: str) -> None:
        raise TypeError(f"`{self.__class__.__name__}.{inspect.stack()[0][3]}` is not supported.")

    async def create_temporary_url(self, file_id: str, expire_after: float) -> str:
        url = self._get_url(self._RETRIEVE_ENDPOINT).format(file_id=file_id)
        headers: Dict[str, str] = {}
        headers.update(self._get_default_headers())
        resp_bytes = await self._request(
            "GET",
            url,
            params={"expirationInSeconds": expire_after},
            headers=headers,
            raise_for_status=True,
        )
        result = self._get_result_from_response_body(resp_bytes)
        return result["fileUrl"]

    async def close(self) -> None:
        if not self._closed:
            await self._session.close()

    def _create_aiohttp_session(self) -> aiohttp.ClientSession:
        return aiohttp.ClientSession(**self._get_session_config())

    def _get_session_config(self) -> Dict[str, Any]:
        return {}

    def _get_default_headers(self) -> Dict[str, str]:
        return {
            "Authorization": f"token {self._access_token}",
        }

    async def _request(self, *args: Any, **kwargs: Any) -> bytes:
        async with self._session.request(*args, **kwargs) as response:
            return await response.read()

    def _create_file_obj_from_dict(self, dict_: Dict[str, Any]) -> RemoteFile:
        metadata: Dict[str, Any]
        if dict_.get("meta"):
            metadata = json.loads(dict_["meta"])
            if not isinstance(metadata, dict):
                raise FileError(f"Invalid metadata: {dict_['meta']}")
        else:
            metadata = {}
        return RemoteFile(
            id=dict_["fileId"],
            filename=dict_["fileName"],
            byte_size=dict_["bytes"],
            created_at=dict_["createTime"],
            purpose=dict_["purpose"],
            metadata=metadata,
            client=self,
        )

    def _get_result_from_response_body(self, resp_body: bytes) -> Any:
        decoded_resp_body = resp_body.decode("utf-8")
        try:
            resp_dict = json.loads(decoded_resp_body)
        except json.JSONDecodeError:
            raise FileError(f"The response body is not valid JSON: {decoded_resp_body}")
        if not isinstance(resp_dict, dict):
            raise FileError(f"The response body can not be parsed as a dict: {decoded_resp_body}")
        if resp_dict.get("errorCode", -1) != 0:
            raise FileError(f"An error was encountered. Response body: {resp_dict}")
        if "result" not in resp_dict:
            raise FileError(f"The response body does not contain the 'result' key: {resp_dict}")
        return resp_dict["result"]

    @classmethod
    def _get_url(cls, path: str) -> str:
        base_url = os.getenv("AISTUDIO_BASE_URL", cls._BASE_URL)
        return f"{base_url}{path}"

__init__ 

__init__(access_token: str, *, aiohttp_session: Optional[aiohttp.ClientSession] = None) -> None

Initialize the AIStudioFileClient.

Parameters:

Name Type Description Default
access_token str

The access token for AI Studio.

 required
aiohttp_session Optional[ClientSession]

A custom aiohttp session (default is None).

 None
Source code in erniebot-agent/src/erniebot_agent/file/remote_file.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
def __init__(
    self, access_token: str, *, aiohttp_session: Optional[aiohttp.ClientSession] = None
) -> None:
    """
    Initialize the AIStudioFileClient.

    Args:
        access_token (str): The access token for AI Studio.
        aiohttp_session (Optional[aiohttp.ClientSession]): A custom aiohttp session (default is None).

    """
    super().__init__()
    self._access_token = access_token
    if aiohttp_session is None:
        aiohttp_session = self._create_aiohttp_session()
    self._session = aiohttp_session
    self._closed = False

list_files async 

list_files() -> List[RemoteFile]

List files in AI Studio client.

Source code in erniebot-agent/src/erniebot_agent/file/remote_file.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
async def list_files(self) -> List[RemoteFile]:
    """List files in AI Studio client."""
    self.ensure_not_closed()
    url = self._get_url(self._LIST_ENDPOINT)
    headers: Dict[str, str] = {}
    headers.update(self._get_default_headers())
    resp_bytes = await self._request(
        "GET",
        url,
        headers=headers,
        raise_for_status=True,
    )
    result = self._get_result_from_response_body(resp_bytes)
    files: List[RemoteFile] = []
    for item in result:
        file = self._create_file_obj_from_dict(item)
        files.append(file)
    return files

retrieve_file async 

retrieve_file(file_id: str) -> RemoteFile

Retrieve a file in AI Studio client by id.

Source code in erniebot-agent/src/erniebot_agent/file/remote_file.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
async def retrieve_file(self, file_id: str) -> RemoteFile:
    """Retrieve a file in AI Studio client by id."""
    self.ensure_not_closed()
    url = self._get_url(self._RETRIEVE_ENDPOINT).format(file_id=file_id)
    headers: Dict[str, str] = {}
    headers.update(self._get_default_headers())
    resp_bytes = await self._request(
        "GET",
        url,
        headers=headers,
        raise_for_status=True,
    )
    result = self._get_result_from_response_body(resp_bytes)
    return self._create_file_obj_from_dict(result)

retrieve_file_contents async 

retrieve_file_contents(file_id: str) -> bytes

Retrieve file content in AI Studio client by id.

Source code in erniebot-agent/src/erniebot_agent/file/remote_file.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
async def retrieve_file_contents(self, file_id: str) -> bytes:
    """Retrieve file content in AI Studio client by id."""
    self.ensure_not_closed()
    url = self._get_url(self._RETRIEVE_CONTENTS_ENDPOINT).format(file_id=file_id)
    headers: Dict[str, str] = {}
    headers.update(self._get_default_headers())
    resp_bytes = await self._request(
        "GET",
        url,
        headers=headers,
        raise_for_status=True,
    )
    return resp_bytes

upload_file async 

upload_file(file_path: pathlib.Path, file_purpose: protocol.FilePurpose, file_metadata: Dict[str, Any]) -> RemoteFile

Upload a file to AI Studio client.

Source code in erniebot-agent/src/erniebot_agent/file/remote_file.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
async def upload_file(
    self, file_path: pathlib.Path, file_purpose: protocol.FilePurpose, file_metadata: Dict[str, Any]
) -> RemoteFile:
    """Upload a file to AI Studio client."""
    self.ensure_not_closed()
    url = self._get_url(self._UPLOAD_ENDPOINT)
    headers: Dict[str, str] = {}
    headers.update(self._get_default_headers())
    with file_path.open("rb") as file:
        form_data = aiohttp.FormData()
        form_data.add_field("file", file, filename=file_path.name)
        form_data.add_field("purpose", file_purpose)
        form_data.add_field("meta", json.dumps(file_metadata))
        resp_bytes = await self._request(
            "POST",
            url,
            data=form_data,
            headers=headers,
            raise_for_status=True,
        )
    result = self._get_result_from_response_body(resp_bytes)
    return self._create_file_obj_from_dict(result)