AI Agents · 课时

带错误处理的安全文件操作

Q: 「带错误处理的安全文件操作」这节课中我会学到什么？

检查文件是否存在和权限，并妥善处理 IO 错误。 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

检查文件是否存在和权限，并妥善处理 IO 错误。

带错误处理的安全文件操作 是 CoddyKit 上的免费 AI Agents 课时。这是第 4 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

安全文件操作为何重要

智能体中的文件操作可能以许多方式失败：文件不存在、智能体没有权限、路径实际是一个目录，或者磁盘空间在写入过程中耗尽。如果智能体因文件错误而崩溃，就会留下不完整的输出和损坏的状态。通过适当的检查和错误处理进行防御式编程，可以让智能体更具韧性。

from pathlib import Path

# Unsafe: crashes with FileNotFoundError
# content = Path('missing.txt').read_text()

# Safe: check first
path = Path('config.json')
if path.exists():
    content = path.read_text(encoding='utf-8')
    print('Loaded config')
else:
    print(f'Config not found at {path.resolve()}')
    content = '{}'  # use default

Path.exists() 和 Path.is_file()

读取文件之前，请检查文件是否存在，并且确实是普通文件（而不是目录、指向目录的符号链接或特殊文件）。Path.exists() 对任何文件系统对象都会返回 True；Path.is_file() 仅对普通文件返回 True。

from pathlib import Path

path = Path('data/report.csv')

# Chain of checks
if not path.exists():
    print(f'Not found: {path}')
elif not path.is_file():
    print(f'Not a regular file: {path} (is_dir={path.is_dir()})')
elif path.stat().st_size == 0:
    print(f'File is empty: {path}')
else:
    # Safe to read
    import csv
    with open(path, 'r', encoding='utf-8', newline='') as f:
        reader = csv.DictReader(f)
        rows = list(reader)
    print(f'Read {len(rows)} rows')

os.access() — 检查权限

os.access(path, mode) 检查当前进程是否拥有文件的指定权限。读取使用 os.R_OK，写入使用 os.W_OK，执行使用 os.X_OK。在尝试需要特定权限的操作之前，这项检查非常有用。

import os
from pathlib import Path

def check_file_access(path):
    p = Path(path)

    checks = {
        'exists': p.exists(),
        'is_file': p.is_file(),
        'readable': os.access(p, os.R_OK),
        'writable': os.access(p, os.W_OK),
    }

    for check, result in checks.items():
        status = 'OK' if result else 'FAIL'
        print(f'  {check}: {status}')

    return all(checks.values())

if check_file_access('data/input.json'):
    print('File is accessible')
else:
    print('Access problem — check path and permissions')

捕获 FileNotFoundError

当您尝试打开不存在的文件时，会引发 FileNotFoundError（OSError 的子类）。请显式捕获该错误，并提供包含预期路径的有用错误消息，以便用户或操作人员准确知道缺少什么。

import json
from pathlib import Path

def load_agent_config(config_path='agent_config.json'):
    path = Path(config_path)
    try:
        with open(path, 'r', encoding='utf-8') as f:
            return json.load(f)
    except FileNotFoundError:
        print(f'Config file not found: {path.resolve()}')
        print('Create agent_config.json with your settings')
        print('Example: {"model": "gpt-4o", "max_retries": 3}')
        return {}  # return empty config as default
    except json.JSONDecodeError as e:
        print(f'Invalid JSON in {path}: {e}')
        return {}

# --- demo ---
config = load_agent_config('does_not_exist_config.json')
print(f'Config used: {config}')

捕获 PermissionError

当进程没有读取或写入文件的权限时，会引发 PermissionError。系统文件、其他用户拥有的文件或权限受限的文件都可能导致此错误。请始终将它与 FileNotFoundError 分开捕获，因为两者需要不同的处理方式。

from pathlib import Path

def read_file_safely(path):
    try:
        return Path(path).read_text(encoding='utf-8')
    except FileNotFoundError:
        print(f'File not found: {path}')
        return None
    except PermissionError:
        import os
        print(f'Permission denied: {path}')
        print(f'File permissions: {oct(Path(path).stat().st_mode)}')
        print(f'Current user: {os.getlogin()}')
        print('Try: chmod +r ' + str(path))
        return None
    except IsADirectoryError:
        print(f'Path is a directory, not a file: {path}')
        return None

# --- demo ---
import os
print(read_file_safely('does_not_exist.txt'))

os.makedirs('a_directory', exist_ok=True)
print(read_file_safely('a_directory'))

捕获 IsADirectoryError

当您尝试将目录当作文件打开时，会引发 IsADirectoryError。当智能体错误地构造路径时，就可能发生这种情况，例如追加的文件名实际上已经是一个目录。请始终捕获此错误，以输出有意义的错误信息。

from pathlib import Path

def safe_write(output_path, content):
    path = Path(output_path)

    # Check the path is not an existing directory
    if path.is_dir():
        raise IsADirectoryError(
            f'Cannot write file: {path} is a directory. '
            f'Use a filename like {path}/output.txt instead.'
        )

    # Ensure parent directory exists
    path.parent.mkdir(parents=True, exist_ok=True)

    try:
        path.write_text(content, encoding='utf-8')
        print(f'Written: {path} ({len(content)} chars)')
    except IsADirectoryError as e:
        print(f'Path error: {e}')
    except PermissionError:
        print(f'Cannot write to {path} — permission denied')

# --- demo ---
import os
safe_write('demo_output/report.txt', 'Agent finished the task.')

os.makedirs('already_a_dir', exist_ok=True)
try:
    safe_write('already_a_dir', 'this will fail')
except IsADirectoryError as e:
    print(f'Rejected: {e}')

使用 tempfile 进行原子写入

直接写入文件很危险——如果智能体在写入过程中崩溃，文件就会只写入一部分并损坏。解决方案是执行原子写入：先写入临时文件，然后将其重命名为最终路径。在 POSIX 系统上，重命名是原子的——最终文件要么是旧版本，要么是新版本，绝不会是残缺文件。

import tempfile
import os
import json
from pathlib import Path

def atomic_write_json(file_path, data):
    path = Path(file_path)
    path.parent.mkdir(parents=True, exist_ok=True)

    # Write to temp file in same directory
    tmp_fd, tmp_path = tempfile.mkstemp(
        dir=path.parent,
        prefix='.tmp_',
        suffix='.json'
    )
    try:
        with os.fdopen(tmp_fd, 'w', encoding='utf-8') as f:
            json.dump(data, f, indent=2, ensure_ascii=False)
        # Atomic rename: replaces final file in one operation
        os.replace(tmp_path, path)
        print(f'Atomically wrote: {path}')
    except Exception as e:
        os.unlink(tmp_path)  # clean up temp file on error
        raise

# --- demo ---
atomic_write_json('demo_state/state.json', {'step': 3, 'status': 'running'})
print('File contents:', Path('demo_state/state.json').read_text(encoding='utf-8'))

并发智能体的文件锁定

当多个智能体实例并行运行并写入同一个文件时，竞态条件会导致数据损坏。请使用文件锁定，借助 fcntl 模块（Linux/macOS）或跨平台的 filelock 库来防止并发写入。

from filelock import FileLock, Timeout
import json
from pathlib import Path

COUNTER_FILE = Path('shared_counter.json')
LOCK_FILE = Path('shared_counter.json.lock')

def increment_counter():
    lock = FileLock(str(LOCK_FILE), timeout=10)
    try:
        with lock:
            # Only one process can be here at a time
            if COUNTER_FILE.exists():
                data = json.loads(COUNTER_FILE.read_text())
            else:
                data = {'count': 0}

            data['count'] += 1
            COUNTER_FILE.write_text(
                json.dumps(data, indent=2)
            )
            return data['count']
    except Timeout:
        print('Could not acquire lock within 10 seconds')
        return None

通过存在性检查安全删除

删除不存在的文件会引发 FileNotFoundError。使用 Path.unlink() 删除目录会引发 IsADirectoryError。请使用 Path.unlink(missing_ok=True)（Python 3.8 及更高版本），或先检查文件是否存在，以实现安全删除。

from pathlib import Path
import shutil

# Safe file deletion (Python 3.8+)
Path('temp_output.json').unlink(missing_ok=True)

# Safe directory deletion
def safe_remove(path):
    p = Path(path)
    if not p.exists():
        print(f'Already gone: {p}')
        return
    if p.is_file():
        p.unlink()
        print(f'Deleted file: {p}')
    elif p.is_dir():
        shutil.rmtree(p)
        print(f'Deleted directory: {p}')
    else:
        print(f'Unknown file type: {p}')

# Clean up temporary workspace
safe_remove('/tmp/agent_workspace/run_001')

写入前检查磁盘空间

写入大型文件（AI 输出、数据集、日志）的智能体应先检查可用磁盘空间。shutil.disk_usage(path) 会返回总字节数、已用字节数和可用字节数。在开始长时间写入操作之前，请检查可用空间是否超过预期输出大小。

import shutil
from pathlib import Path

def check_disk_space(output_dir, required_bytes):
    path = Path(output_dir)
    path.mkdir(parents=True, exist_ok=True)

    usage = shutil.disk_usage(path)
    free_gb = usage.free / (1024 ** 3)
    required_gb = required_bytes / (1024 ** 3)

    print(f'Disk free: {free_gb:.2f} GB')
    print(f'Required:  {required_gb:.2f} GB')

    if usage.free < required_bytes * 1.1:  # 10% safety margin
        raise IOError(
            f'Insufficient disk space: '
            f'{free_gb:.2f} GB free, '
            f'{required_gb:.2f} GB required'
        )
    return True

# Before writing a 500 MB dataset
check_disk_space('/tmp/output', 500 * 1024 * 1024)

覆盖前创建备份

当智能体更新现有文件时，最好保留之前版本的备份。如果新内容有误，这样便可以回滚。请使用带时间戳的备份文件名，并限制备份数量，以免占满磁盘空间。

import shutil
import datetime
from pathlib import Path

def write_with_backup(file_path, content, max_backups=5):
    path = Path(file_path)
    backup_dir = path.parent / '.backups'
    backup_dir.mkdir(exist_ok=True)

    # Backup existing file
    if path.exists():
        timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
        backup = backup_dir / f'{path.name}.{timestamp}'
        shutil.copy2(path, backup)
        print(f'Backed up to: {backup}')

    # Write new content
    path.write_text(content, encoding='utf-8')

    # Prune old backups
    backups = sorted(backup_dir.glob(f'{path.name}.*'))
    for old in backups[:-max_backups]:
        old.unlink()
        print(f'Pruned old backup: {old.name}')

# --- demo ---
write_with_backup('demo_notes.txt', 'version 1')
write_with_backup('demo_notes.txt', 'version 2')
print('Backups:', [p.name for p in sorted(Path('.backups').glob('demo_notes.txt.*'))])

快速检查：原子写入

检验您对安全文件写入模式的理解。

安全文件操作回顾

现在，您的智能体已经能够以防御性方式处理文件：

访问前先检查：Path.exists() + Path.is_file() + os.access(path, os.R_OK)
捕获 FileNotFoundError、PermissionError 和 IsADirectoryError，并提供清晰的消息
使用原子写入（临时文件 + os.replace），防止崩溃时产生残缺或损坏的文件
当多个智能体写入同一个文件时，使用文件锁定（filelock 库）
使用 missing_ok=True 安全删除文件，避免 FileNotFoundError
大型写入前检查磁盘空间；覆盖重要文件前保留备份

常见问题解答

「带错误处理的安全文件操作」课时是免费的吗？

是的 — 「带错误处理的安全文件操作」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「带错误处理的安全文件操作」这节课中我会学到什么？

检查文件是否存在和权限，并妥善处理 IO 错误。你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 4 节课，共 4 节。

「带错误处理的安全文件操作」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 AI Agents